package com.shujia.flink.core

import java.time.Duration

import com.shujia.flink.core.Demo3EventTime.Event
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, SlidingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object Demo4TImeWindow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    //指定时间模式为事件时间
    //默认是处理时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)


    val events: DataStream[Event] = lineDS.map(line => Event(line.split(",")(0), line.split(",")(1).toLong))


    /**
      * 时间窗口
      * SlidingProcessingTimeWindows： 处理时间的滑动窗口
      * SlidingEventTimeWindows ： 事件事件的滑动窗口
      * TumblingProcessingTimeWindows : 处理事件的滚动窗口
      * TumblingEventTimeWindows： 事件事件的滚动窗口
      *
      * 如果要使用事件事件需要指定时间模式，时间字段，水位线
      *
      */
    events
      //设置水位线和时间字段
      .assignTimestampsAndWatermarks( //指定数据最大的延迟时间
      new BoundedOutOfOrdernessTimestampExtractor[Event](Time.seconds(5)) {
        //返回事件时间字段
        override def extractTimestamp(element: Event): Long = {
          element.ts
        }
      })
      .map(event => (event.id, 1))
      .keyBy(_._1)
      .window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)))
      .sum(1)
      .print()

    env.execute()

  }
}
